#include "skynet.h"
#include "socket_server.h"
#include "skynet_malloc.h"
#include "atomic.h"
#include <string.h>
#include <errno.h>
#include <stdio.h>
#include <signal.h>

#include <event2/bufferevent.h>
#include <event2/buffer.h>
#include <event2/listener.h>
#include <event2/util.h>
#include <event2/event.h>
#include <event2/event_struct.h>
#include <event2/bufferevent_struct.h>


#define MAX_INFO 128
// MAX_SOCKET will be 2^MAX_SOCKET_P
#define MAX_SOCKET_P 16
#define MAX_EVENT 64
#define MIN_READ_BUFFER 64
#define SOCKET_TYPE_INVALID 0
#define SOCKET_TYPE_RESERVE 1
#define SOCKET_TYPE_PLISTEN 2
#define SOCKET_TYPE_LISTEN 3
#define SOCKET_TYPE_CONNECTING 4
#define SOCKET_TYPE_CONNECTED 5
#define SOCKET_TYPE_HALFCLOSE 6
#define SOCKET_TYPE_PACCEPT 7
#define SOCKET_TYPE_BIND 8

#define MAX_SOCKET (1<<MAX_SOCKET_P)


#define HASH_ID(id) (((unsigned)id) % MAX_SOCKET)

struct socket {
	uintptr_t opaque;
	int fd;
	int id;
	uint16_t type;
	void* dataproxy;
	struct socket_server* server;
};

struct socket_server
{
	evutil_socket_t  recvctrl_fd;
	evutil_socket_t sendctrl_fd;
	int alloc_id;
	struct event_base* base;
	struct socket slot[MAX_SOCKET];
	SendCallback sendcallback;
	char buffer[256];
};

struct request_open {
	int id;
	int port;
	uintptr_t opaque;
	char host[1];
};

struct request_send {
	int id;
	int sz;
	char * buffer;
};

struct request_close {
	int id;
	int shutdown;
	uintptr_t opaque;
};

struct request_listen {
	int id;
	uintptr_t opaque;
	int port;
	int backlog;
	evutil_socket_t fd;
	char host[1];
};

struct request_bind {
	int id;
	int fd;
	uintptr_t opaque;
};

struct request_start {
	int id;
	uintptr_t opaque;
};

struct request_setopt {
	int id;
	int what;
	int value;
};

struct request_package {
	uint8_t header[8];	// 6 bytes dummy
	union {
		char buffer[256];
		struct request_open open;
		struct request_send send;
		struct request_close close;
		struct request_listen listen;
		struct request_bind bind;
		struct request_start start;
		struct request_setopt setopt;
	} u;
	uint8_t dummy[256];
};

struct send_object {
	void * buffer;
	int sz;
	void(*free_func)(void *);
};

static void innerReadCallback(int sockfd, short ievent, void* arg);
static int block_readinner(int, struct evbuffer*,size_t howmuch);
static int listen_socket(struct socket_server*, struct request_listen*, struct socket_message*);
static int start_socket(struct socket_server*, struct request_start*, struct socket_message*);
static int bind_socket(struct socket_server*, struct request_bind*, struct socket_message*);
static void send_socket(struct socket_server*, struct request_send*);
static int reserve_id(struct socket_server*);
static void accept_conn_cb(struct evconnlistener* listener, evutil_socket_t fd, struct sockaddr* address, int socklen, void* ctx);
static void write_cb(struct bufferevent* bev, void* ctx);
static void read_cb(struct bufferevent* bev, void* ctx);
static void event_cb(struct bufferevent* bev, short events, void* ctx);
static void force_close(struct socket_server*, struct socket*, struct socket_message*);
static void send_request(struct socket_server*, struct request_package*,char,int);
static int open_request(struct socket_server*, struct request_package*, uintptr_t opaque, const char* addr, int port);
static int reverse_id(struct socket_server*);
static struct socket* new_fd(struct socket_server*, size_t id,size_t fd , uintptr_t opaque);
struct socket_server * socket_server_create(SendCallback sendcallback)
{
	struct event_base* base = event_base_new();
	if (base == 0)
	{
		return 0;
	}

	struct socket_server* server = (struct socket_server*)skynet_malloc(sizeof(struct socket_server));
	if (server == 0)
	{
		event_base_free(base);
		return 0;
	}
	evutil_socket_t pairs[2];
	if (evutil_socketpair(AF_INET, SOCK_STREAM, 0, pairs) == -1)
	{
		event_base_free(base);
		skynet_free(server);
		return 0;
	}
	server->recvctrl_fd = pairs[0];
	server->sendctrl_fd = pairs[1];

	struct event ev;
	event_assign(&ev, base, server->recvctrl_fd, EV_PERSIST | EV_READ, innerReadCallback, server);
	event_add(&ev, NULL);

	int i;
	for (i = 0; i < MAX_SOCKET; i++) {
		struct socket *s = &server->slot[i];
		s->type = SOCKET_TYPE_INVALID;
		s->server = server;
	}
	server->alloc_id = 0;
	server->sendcallback = sendcallback;
	return server;

}

void socket_server_release(struct socket_server* server)
{
	evutil_closesocket(server->recvctrl_fd);
	evutil_closesocket(server->sendctrl_fd);
	event_base_free(server->base);
	int i;
	struct socket_message dummy;
	for (i = 0; i < MAX_SOCKET; i++) {
		struct socket *s = &server->slot[i];
		if (s->type != SOCKET_TYPE_RESERVE) {
			force_close(server, s, &dummy);
		}
	}
	skynet_free(server);
}

void socket_server_exit(struct socket_server * server)
{
	struct request_package request;
	send_request(server, &request, 'X', 0);
}

void socket_server_shutdown(struct socket_server *server, uintptr_t opaque, int id)
{
	struct request_package request;
	request.u.close.id = id;
	request.u.close.shutdown = 1;
	request.u.close.opaque = opaque;
	send_request(server, &request, 'K', sizeof(request.u.close));
}
void socket_server_start(struct socket_server *server, uintptr_t opaque, int id)
{
	struct request_package request;
	request.u.start.id = id;
	request.u.start.opaque = opaque;
	send_request(server, &request, 'S', sizeof(request.u.start));
}

int64_t socket_server_send(struct socket_server *server, int id, const void * buffer, int sz)
{
	struct socket * s = &server->slot[HASH_ID(id)];
	if (s->id != id || s->type == SOCKET_TYPE_INVALID) {
		///free_buffer(server, buffer, sz);
		return -1;
	}

	struct request_package request;
	request.u.send.id = id;
	request.u.send.sz = sz;
	request.u.send.buffer = (char *)buffer;

	send_request(server, &request, 'D', sizeof(request.u.send));
	return 0;
}

// ctrl command below returns id
int socket_server_listen(struct socket_server *server, uintptr_t opaque, const char * addr, int port, int backlog)
{
	int id = reserve_id(server);
	if (id < 0)
	{
		return id;
	}
	struct request_package request;
	request.u.listen.opaque = opaque;
	request.u.listen.id = id;
	request.u.listen.port = port;
	request.u.listen.backlog = backlog;
	if (addr)
	{
		memcpy((void*)&request.u.listen.host[0], addr, strlen(addr));
	}
	else
	{
		request.u.listen.host[0] = '0';
	}

	send_request(server, &request, 'L', sizeof(request.u.listen));
	return id;
}
int socket_server_connect(struct socket_server *server, uintptr_t opaque, const char * addr, int port)
{
	struct request_package request;
	int len = open_request(server, &request, opaque, addr, port);
	if (len < 0)
		return -1;
	send_request(server, &request, 'O', sizeof(request.u.open) + len);
	return request.u.open.id;
}
int socket_server_bind(struct socket_server *server, uintptr_t opaque, int fd)
{
	struct request_package request;
	int id = reserve_id(server);
	if (id < 0)
		return -1;
	request.u.bind.opaque = opaque;
	request.u.bind.id = id;
	request.u.bind.fd = fd;
	send_request(server, &request, 'B', sizeof(request.u.bind));
	return id;
}

// for tcp
void socket_server_nodelay(struct socket_server *server, int id)
{
	struct request_package request;
	request.u.setopt.id = id;
	request.u.setopt.what = TCP_NODELAY;
	request.u.setopt.value = 1;
	send_request(server, &request, 'T', sizeof(request.u.setopt));
}

void
socket_server_close(struct socket_server *ss, uintptr_t opaque, int id) {
	struct request_package request;
	request.u.close.id = id;
	request.u.close.shutdown = 0;
	request.u.close.opaque = opaque;
	send_request(ss, &request, 'K', sizeof(request.u.close));
}

//////////////////////////////////////////////////////////////////////////////////////
int block_readinner(evutil_socket_t sockfd, struct evbuffer* buffer,size_t howmuch)
{
	for (;;)
	{
		size_t n = evbuffer_read(buffer, sockfd, howmuch);
		if (n < 0)
		{
			if (errno == EINTR)
			{
				continue;
			}
			return 0;
		}
		return n;
	}
}
void innerReadCallback(int sockfd, short ievent, void* arg)
{
	struct evbuffer* header = evbuffer_new();
	uint8_t headerflag[2];
	block_readinner(sockfd, header, sizeof(headerflag));
	evbuffer_remove(header, headerflag, sizeof(headerflag));
	uint8_t type = headerflag[0];
	uint8_t len = headerflag[1];
	struct evbuffer* column = evbuffer_new();
	int reallen = block_readinner(sockfd, column, len);
	uint8_t buffer[256];
	evbuffer_remove(column, buffer, reallen);
	
	struct socket_server* ss = (struct socket_server*)arg;
	struct socket_message result;
	int datatype;
	int more = 0;
	switch (type)
	{
	case 'L':
		datatype = listen_socket(ss, (struct request_listen*)buffer, &result);
		if (datatype != 0 && ss->sendcallback)
		{
			ss->sendcallback(ss, datatype, &result, more);
		}
		break;

	case 'S':
		datatype = start_socket(ss, (struct request_start*)buffer, &result);
		if (datatype != 0 && ss->sendcallback)
		{
			ss->sendcallback(ss, datatype, &result, more);
		}
		break;
	case 'B':
		datatype = bind_socket(ss, (struct request_bind*)buffer, &result);
		if (datatype != 0 && ss->sendcallback)
		{
			ss->sendcallback(ss, datatype, &result, more);
		}
		break;
	case 'D':
		send_socket(ss, (struct request_send*)buffer);
		break;
	}
	evbuffer_free(header);
	evbuffer_free(column);
}

int reserve_id(struct socket_server* ss)
{
	int i;
	for (i = 0; i < MAX_SOCKET; i++)
	{
		int id = ss->alloc_id;
		if (id < 0)
		{
			id = ss->alloc_id & 0x7fffffff;
		}
		struct socket* s = &ss->slot[HASH_ID(id)];
		if (s->type == SOCKET_TYPE_INVALID)
		{
			s->type = SOCKET_TYPE_RESERVE;
			s->id = id;
			s->fd = -1;
			return id;
		}
	}
	return -1;
}
int listen_socket(struct socket_server* ss, struct request_listen* request, struct socket_message* message)
{
	struct socket* sock = &ss->slot[HASH_ID(request->id)];
	if (sock->type != SOCKET_TYPE_RESERVE)
	{
		return -1;
	}

	struct evconnlistener* listener;
	struct event_base* base = ss->base;
	struct evutil_addrinfo ai_hints;
	struct evutil_addrinfo* ai_answer = NULL;
	char portstr[16];
	char* addr;
	if (request->host[0] == '0')
	{
		addr = "0.0.0.0";
	}
	else {
		addr = request->host;
	}
	sprintf(portstr, "%d", request->port);
	memset(&ai_hints, 0, sizeof(ai_hints));
	ai_hints.ai_family = AF_UNSPEC;
	ai_hints.ai_socktype = SOCK_STREAM;
	ai_hints.ai_protocol = IPPROTO_TCP;
	int status = evutil_getaddrinfo(addr,portstr,&ai_hints,&ai_answer);
	if (status != 0)
	{
		fprintf(stderr, "Error [%s:%s] err:%s", addr, portstr, evutil_gai_strerror(status));
		return -1;
	}

	listener = evconnlistener_new_bind(base, NULL, NULL, LEV_OPT_CLOSE_ON_FREE | LEV_OPT_REUSEABLE, -1, (struct sockaddr*)ai_answer->ai_addr, ai_answer->ai_addrlen);
	if (listener == NULL)
	{
		return -1;
	}
	evutil_socket_t fd = evconnlistener_get_fd(listener);
	sock->fd = fd;
	sock->type = SOCKET_TYPE_PLISTEN;
	sock->dataproxy = (void*)listener;
	return -1;
}

void accept_conn_cb(struct evconnlistener* listener, evutil_socket_t fd, struct sockaddr* address, int socklen, void* ctx)
{
	struct socket* socket = (struct socket*)ctx;
	struct socket_server* ss = socket->server;
	struct event_base* base = ss->base;
	int id = reverse_id(ss);
	if (id < 0)
	{
		evutil_closesocket(fd);
		return;
	}
	struct socket* ns = new_fd(ss, id, fd, socket->opaque);
	if (ns == NULL)
	{
		evutil_closesocket(fd);
		return;
	}
	ns->type = SOCKET_TYPE_PACCEPT;
	struct socket_message result;
	result.opaque = socket->opaque;
	result.id = socket->id;
	result.ud = id;
	result.data = NULL;
	struct sockaddr_in* addr = (struct sockaddr_in*)address;
	int port = ntohs(addr->sin_port);
	char tmp[128];
	evutil_inet_ntop(addr->sin_family, (void*)&addr->sin_addr, tmp, sizeof(tmp));
	sprintf(ss->buffer, "%s:%d", tmp, port);;
	result.data = ss->buffer;
	if (ss->sendcallback)
	{
		ss->sendcallback(ss,SOCKET_ACCEPT, &result, 0);
	}
	return;
}

int start_socket(struct socket_server* ss, struct request_start* request, struct socket_message* result)
{
	int id = request->id;
	result->id = id;
	result->opaque = request->opaque;
	result->ud = 0;
	result->data = NULL;
	struct socket* socket = &ss->slot[HASH_ID(id)];
	if (socket->type == SOCKET_TYPE_PLISTEN)
	{
		socket->type = SOCKET_TYPE_LISTEN;
		socket->opaque = request->opaque;
		result->data = "start";
		struct evconnlistener* listener = (struct evconnlistener*)socket->dataproxy;
		evconnlistener_set_cb(listener, accept_conn_cb,(void*)socket);
		evconnlistener_enable(listener);
		return SOCKET_OPEN;
	}
	if (socket->type == SOCKET_TYPE_PACCEPT)
	{
		socket->type = SOCKET_TYPE_CONNECTED;
		socket->opaque = request->opaque;
		result->data = "start";
		struct event_base * base = ss->base;
		struct bufferevent* bev = bufferevent_socket_new(base, socket->fd, BEV_OPT_CLOSE_ON_FREE);
		socket->dataproxy = bev;
		bufferevent_setcb(bev, read_cb, write_cb, event_cb, (void*)socket);
		bufferevent_enable(bev, EV_READ | EV_WRITE);
		return SOCKET_OPEN;
	}
	return -1;
}

void write_cb(struct bufferevent* bev, void* ctx)
{
	if (evbuffer_get_length(bev->output) == 0)
	{
		evbuffer_drain(bev->output, evbuffer_get_length(bev->output));
	}
	
}
void read_cb(struct bufferevent* bev, void* ctx)
{
	struct evbuffer* input = bufferevent_get_input(bev);
	struct socket* socket = (struct socket*)ctx;
	if (input)
	{
		size_t nz = evbuffer_get_length(input);
		if (nz > 0)
		{
			char* data = (char*)skynet_malloc(nz + 1);
			data[nz] = '0';
			evbuffer_remove(input, data, nz);
			struct socket_message result;
			result.opaque = socket->opaque;
			result.id = socket->id;
			result.ud = nz;
			result.data = data;
			if (socket->server->sendcallback)
			{
				socket->server->sendcallback(socket->server, SOCKET_DATA, &result, 0);
			}
		}
	}
	return;
}

void event_cb(struct bufferevent* bev, short events, void* ctx)
{
	if ((events &  BEV_EVENT_EOF) || (events & BEV_EVENT_ERROR))
	{
		struct event_base* base = bufferevent_get_base(bev);
		struct socket* socket = (struct socket*)ctx;
		evutil_socket_t fd = bufferevent_getfd(bev);
		bufferevent_free(bev);
		evutil_closesocket(fd);
		if (socket->server->sendcallback)
		{
			struct socket_message result;
			result.id = socket->id;
			result.ud = 0;
			result.data = NULL;
			result.opaque = socket->opaque;
			if (socket->type == SOCKET_TYPE_INVALID)
			{
				return;
			}
			socket->type = SOCKET_TYPE_INVALID;
			socket->server->sendcallback(socket->server, SOCKET_CLOSE, &result, 0);
		}
	}
}

void send_socket(struct socket_server* server, struct request_send* request)
{
	int id = request->id;
	struct socket* s = &server->slot[HASH_ID(id)];
	if (s->type != SOCKET_TYPE_CONNECTED)
	{
		return;
	}
	struct bufferevent* bev = (struct bufferevent*)s->dataproxy;
	struct evbuffer* output = bufferevent_get_output(bev);
	evbuffer_add(output, request->buffer, request->sz);
	evbuffer_unfreeze(output, 1);
	evbuffer_write(output, s->fd);
	skynet_free(request->buffer);
}
int bind_socket(struct socket_server* socketserver, struct request_bind* request, struct socket_message* result)
{
	return 1;
}

void force_close(struct socket_server* server, struct socket* socket, struct socket_message* message)
{

}
void send_request(struct socket_server* server, struct request_package* package, char flag, int len)
{

}

int open_request(struct socket_server* server, struct request_package* package, uintptr_t opaque, const char* addr, int port)
{
	return 0;
}

int reverse_id(struct socket_server* ss)
{
	int i;
	for (i = 0; i< MAX_SOCKET; i++)
	{
		int id = ATOM_INC(&(ss->alloc_id));
		if (id < 0)
		{
			id = ATOM_AND(&(ss->alloc_id), 0x7fffffff);
		}
		struct socket* s = &ss->slot[HASH_ID(id)];
		if (s->type == SOCKET_TYPE_INVALID)
		{
			if (ATOM_CAS((long*)&s->type, (long)SOCKET_TYPE_INVALID, (long)SOCKET_TYPE_RESERVE))
			{
				s->id - id; 
				s->fd = -1;
				return id;
			}
			return -1;
			
		}

	}
	return -1;
}


struct socket* new_fd(struct socket_server* ss, size_t id, size_t fd, uintptr_t opaque)
{
	struct socket* s = &ss->slot[HASH_ID(id)];
	s->id = id;
	s->fd = fd;
	s->opaque = opaque;
	return s;
}